/*
 * Copyright 2015 MingxingWang

 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at

 *     http://www.apache.org/licenses/LICENSE-2.0

 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.audaque.graphdb

import java.io._

import com.audaque.graphdb.index.LuceneIndex
import org.apache.spark.graphx._
import org.apache.spark.graphx.graphdb.LEdges

import scala.collection.immutable.Set
import scala.collection.mutable


/**
 * Created by Ming on 2015/3/26.
 */
object LGraph {
  def apply(graphName: String, path: String): LGraph = new LGraph(graphName, path)
}

class LGraph(val graphName: String,
             var path: String,
             val vertices: LVertices = new LVertices(),
             val edges: LEdges = new LEdges()) {

  private var maxVid: VertexId = 0

  val graphRootPath =
    if (path == null || path.isEmpty) {
      new File(graphName)
    } else {
      new File(path)
    }

  val schemaFactory: SchemaFactory = new SchemaFactory()
  var indexer: LuceneIndex = null

  def genNewVid: VertexId = {
    maxVid += 1
    maxVid
  }

  def updateMaxVid(vid: Long) {
    maxVid = vid
  }

  def genSchema(name: String, fields: Array[(String, Type)]): Schema = schemaFactory.genSchema(name, fields)

  def getSchema(name: String): Option[Schema] = schemaFactory.getSchemaByName(name)

  def getSchema(sid: Long): Option[Schema] = schemaFactory.getSchemaById(sid)

  /**
   * 总顶点数
   * @return
   */
  def vertexCount(): Long = vertices.count()

  /**
   * 总边数
   * @return
   */
  def edgeCount(): Long = edges.count()

  /**
   * 查找一个顶点
   * @param vid id
   * @return
   */
  def findVertex(vid: VertexId): Option[Vertex] = vertices.find(vid)

  /**
   * 删除一个顶点。如果非孤立点则失效，使用deleteVertexAndEdges
   * @param vid 待删除顶点的id
   * @return  是否成功删除
   */
  def deleteVertex(vid: VertexId): Boolean = {
    vertices.delete(vid)
  }

  /**
   * 删除一个顶点以及与其连接的所有的边
   * @param vid 待删除顶点的id
   * @return 是否成功删除
   */
  def deleteVertexAndEdges(vid: VertexId): Boolean = {
    findNeighborEdges(vid, EdgeDirection.Either).foreach(e => deleteEdge(e.srcId, e.dstId))
    vertices.delete(vid)
    true
  }

  /**
   * 根据属性增加一个顶点，自动获取id，返回Vertex
   * @param prop 新顶点的属性
   * @return Vertex
   */
  def addVertex(prop: Property): Vertex = {
    val vid = genNewVid
    vertices.add(vid, prop)
    Vertex(vid, prop)
  }

  /**
   * 增加新顶点
   * @param vid 新顶点的id
   * @param prop 新顶点的属性
   * @return  是否成功
   */
  def addVertex(vid: VertexId, prop: Property): Boolean = {
    if (vid > maxVid) maxVid = vid
    vertices.add(vid, prop)
  }

  /**
   * 修改顶点
   * @param vid 待修改顶点的id
   * @param newProp 新属性
   * @return 是否成功
   */
  def modifyVertex(vid: VertexId, newProp: Property): Boolean = {
    vertices.modify(vid, newProp)
  }

  private def findExistVertex(vids: Array[VertexId]): Array[Vertex] = {
    if (vids == null || vids.length == 0) {
      return Array[Vertex]()
    }
    val verts = new mutable.ArrayBuffer[Vertex]()
    vids.foreach(vid => findVertex(vid) match {
      case Some(v) => verts += v
      case None =>
    })
    verts.toArray
  }

  /**
   * 查找匹配value的顶点
   * @param value 匹配的属性值
   * @param fuzzyLookup 是否模糊查找
   * @param retNum 最大返回数量
   * @return 匹配的顶点集合
   */
  def searchVertex(value: String,
                   fuzzyLookup: Boolean = false,
                   retNum: Int = 100): Array[Vertex] = {
    searchVertex2(Set[String](value), false, fuzzyLookup, retNum)
  }

  /**
   * 全字段查询多个关键字
   * @param values  待查询的关键字集合
   * @param logic  logic为真各关键字按照“与”规则查询，否则按照“或”规则查询
   * @param fuzzyLookup  为真进行模糊查询，否则精确查询
   * @return 结果按相似度排序
   */
  def searchVertex2(values: Iterable[String],
                    logic: Boolean = false,
                    fuzzyLookup: Boolean = false,
                    retNum: Int = 100): Array[Vertex] = {
    if (indexer == null) {
      return Array[Vertex]()
    }
    val vids = indexer.valueSearch(values, logic, fuzzyLookup, retNum)
    findExistVertex(vids)
  }

  def searchVertexKV(key: String,
                     value: String,
                     fuzzyLookup: Boolean = false,
                     retNum: Int = 100): Array[Vertex] = {
    searchVertexKV2(Set[(String, String)](key -> value), false, fuzzyLookup, retNum)
  }

  /**
   * 键值对的查找方法
   * @param keysAndValues  需要查询的键值对的集合
   * @param logic            需要查询的键值对按“与”还是按“或”查询
   * @param fuzzyLookup   是否按照模糊查询，为true表示模糊查询，否则为精确查询
   * @param retNum           最多返回多少条查询结果
   * @return
   */
  def searchVertexKV2(keysAndValues: Iterable[(String, String)],
                      logic: Boolean = false,
                      fuzzyLookup: Boolean = false,
                      retNum: Int = 100): Array[Vertex] = {
    if (indexer == null) {
      return Array[Vertex]()
    }
    val vids = indexer.pairSearch(keysAndValues, logic, fuzzyLookup, retNum)
    findExistVertex(vids)
  }

  /**
   * 修改边
   * @param srcId
   * @param dstId
   * @param newProp
   * @return
   */
  def modifyEdge(srcId: VertexId, dstId: VertexId, newProp: Property): Boolean = edges.modify(srcId, dstId, newProp)

  /**
   * 增加一条边
   * @param srcId
   * @param dstId
   * @param prop
   * @return
   */
  def addEdge(srcId: VertexId, dstId: VertexId, prop: Property): Boolean = edges.add(srcId, dstId, prop)

  /**
   * 删除一条边
   * @param srcId
   * @param dstId
   * @return
   */
  def deleteEdge(srcId: VertexId, dstId: VertexId): Boolean = edges.delete(srcId, dstId)

  /**
   * 查找边
   * @param srcId
   * @param dstId
   * @return
   */
  def findEdge(srcId: VertexId, dstId: VertexId): Option[Edge] = edges.find(srcId, dstId)

  /**
   * 查找所有从vid出去且满足filter条件的边
   * @param vid
   * @param filter
   * @return
   */
  def findOutEdges(vid: VertexId, filter: Property => Boolean = prop => true): Set[Edge] = edges.findOutEdges(vid, filter)

  /**
   * 查找所有进入到vid且满足filter条件的边
   * @param vid
   * @param filter
   * @return
   */
  def findInEdges(vid: VertexId, filter: Property => Boolean = prop => true): Set[Edge] = edges.findInEdges(vid, filter)

  def findNeighborEdges(vid: VertexId, dir: EdgeDirection,
                        filter: Property => Boolean = edge => true): Set[Edge] = {
    edges.findNeighborEdges(vid, dir, filter)
  }

  def findOutVertexIds(vid: VertexId,
                       edgeFilter: Property => Boolean = ed => true): Set[VertexId] = {
    edges.findOutVertexIds(vid, edgeFilter)
  }

  def findInVertexIds(vid: VertexId,
                      edgeFilter: Property => Boolean = ed => true): Set[VertexId] = {
    edges.findInVertexIds(vid, edgeFilter)
  }

  /**
   * 查找所有从vid出去且是一阶可达的顶点，且出边满足edgeFilter，顶点满足vertexFilter
   * @param vid
   * @param edgeFilter
   * @param vertexFilter
   * @return
   */
  def findOutVertices(vid: VertexId,
                      edgeFilter: Property => Boolean = ed => true,
                      vertexFilter: Property => Boolean = vert => true): Set[Vertex] = {
    val res = new mutable.HashSet[Vertex]()
    val s = findOutVertexIds(vid, edgeFilter).foreach(vid =>
      findVertex(vid) match {
        case Some(v) => if (v != null && vertexFilter(v.prop)) res += v
        case None =>
      })
    res.toSet
  }

  /**
   * 查找所有进入到vid且是一阶可达的顶点，且出边满足edgeFilter，顶点满足vertexFilter
   * @param vid
   * @param edgeFilter
   * @param vertexFilter
   * @return
   */
  def findInVertices(vid: VertexId,
                     edgeFilter: Property => Boolean = ed => true,
                     vertexFilter: Property => Boolean = vert => true): Set[Vertex] = {
    val res = new mutable.HashSet[Vertex]()
    findInVertexIds(vid, edgeFilter).foreach(vid =>
      findVertex(vid) match {
        case Some(v) => if (v != null && vertexFilter(v.prop)) res += v
        case None =>
      })
    res.toSet
  }

  def saveMetaData(): Unit = {
    val metaFile = new File(graphRootPath, GraphDBConf.MetaDataFile)
    val out: DataOutputStream = new DataOutputStream(new FileOutputStream(metaFile))
    out.writeLong(maxVid)
    schemaFactory.save(out)
    out.flush()
    out.close()
  }

  def loadMetaData() {
    val metaFile = new File(graphRootPath, GraphDBConf.MetaDataFile)
    if (!metaFile.exists()) return
    val in: DataInputStream = new DataInputStream(new FileInputStream(metaFile))
    maxVid = in.readLong()
    schemaFactory.load(in)
    in.close()
  }

  private def saveEdges() {
    val edgeFile = new File(graphRootPath, GraphDBConf.EdgeFile)
    val out: DataOutputStream = new DataOutputStream(new BufferedOutputStream(new FileOutputStream(edgeFile)))
    edges.save(out)
    out.flush()
    out.close()
  }

  private def loadEdges() {
    val t1 = System.currentTimeMillis()
    println("Loading edges... ")
    val edgeFile = new File(graphRootPath, GraphDBConf.EdgeFile)
    if (!edgeFile.exists()) return
    val in: DataInputStream = new DataInputStream(new BufferedInputStream(new FileInputStream(edgeFile)))
    edges.load(in, schemaFactory)
    in.close()
    val t2 = System.currentTimeMillis()
    val te = (t2 - t1) / 100 / 10.0
    println("Down, time used: " + te + "s.")
  }

  private def saveVertices() {
    val vertexFile = new File(graphRootPath, GraphDBConf.VertexFile)
    val out: DataOutputStream = new DataOutputStream(new BufferedOutputStream(new FileOutputStream(vertexFile)))
    vertices.save(out)
    out.flush()
    out.close()
  }

  private def loadVertices() {
    println("Loading vertices... ")
    val t1 = System.currentTimeMillis()
    val vertexFile = new File(graphRootPath, GraphDBConf.VertexFile)
    if (!vertexFile.exists()) return
    val in: DataInputStream = new DataInputStream(new BufferedInputStream(new FileInputStream(vertexFile)))
    vertices.load(in, schemaFactory)
    in.close()
    val t2 = System.currentTimeMillis()
    val tv = (t2 - t1) / 100 / 10.0
    println("Down, time used: " + tv + "s.")
  }

  /**
   * 从磁盘加载图数据内容
   */
  def load(): Unit = {
    loadMetaData()
    loadVertices()
    loadEdges()
    val indexPath = new File(graphRootPath, GraphDBConf.IndexPath)
    if (indexPath.exists()) {
      indexer = new LuceneIndex(indexPath)
      indexer.openIndex(schemaFactory)
    } else {
      indexer = null
    }
  }

  /**
   * 保存内容到磁盘
   */
  def save(): Unit = {
    try {
      graphRootPath.mkdirs()
      saveMetaData()
      saveEdges()
      saveVertices()
      if (indexer != null) {
        indexer.commit()
      }
    } catch {
      case e: Throwable => println(e)
    }

  }

  /**
   * 将缓存的增删记录更新到partition中
   */
  def update() {
    vertices.update()
    edges.update()
  }

  def close() = {
    vertices.clear()
    edges.clear()
  }
}
